/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */

/* 
 * File:   KKSManger.cpp
 * Author: ever
 * 
 * Created on 2018年5月12日, 下午11:05
 */

#include "KKSManger.h"

KKSManger::KKSManger() {
}

KKSManger::KKSManger(const KKSManger &orig) {
}

KKSManger::~KKSManger() {
}

void KKSManger::setKKSFilePath(const string &kks_file_path, bool isIndex) {
    _kfile_path = kks_file_path;
    TL_Common::mkdirRecursive(_kfile_path + "/kks/");

    if (isIndex) {
        //是否需要索引文件
        _kks_index.attach(kks_file_path + "/kks/kks.idx");
    }
    time_t now = time(0) + BEIJING_TIME;
    now = now / ONE_DAY_SECONDS * ONE_DAY_SECONDS;
    _kks_file[now] = std::shared_ptr<KKSFile>(new KKSFile());
    _kks_file[now]->attach(getKKSPath(now));
}

KKSIndex &KKSManger::getKKSIndex() {
    return _kks_index;
}

string KKSManger::getKKSPath(time_t t) {
    tidp::TL_Datetime dt(t);
    return _kfile_path + dt.toString("/kks/%Y%m%d.krd"); //kks raw data
}

u32_t KKSManger::addKKS(const string &kks) {
    std::lock_guard<std::mutex> guard(_mux);
    return _kks_index.addKKS(kks);
}

u32_t KKSManger::getKKSId(const string &kks) {
    return _kks_index.getKKSId(kks);
}

void KKSManger::saveIndex() {
    std::lock_guard<std::mutex> guard(_mux);
    _kks_index.save();
}

void KKSManger::addKKSValue(u32_t id, u32_t tm, u32_t vl, u8_t quality) {
    std::lock_guard<std::mutex> guard(_mux);

    //map<time_t,deque<KKS_Value> >
    //记录最近的几个点,用于高速查询与去重
    //TODO: 创建线程去更新点,而不是一个个去更新;
    static KKS_Value v;
    static u32_t jjj = 0;
    auto &dq = _kks_value_cache[id];
    auto tail = dq.rbegin();
    //dq size == 0 的时候,可能是重启,并没有从kks中获取最后一个时间与值,这种情况下可能重复,不去关注
    //tm 在递增的情况下才能插入值,否则自动忽略
    if (dq.size() == 0 || tail->tm < tm) {
        v.quality = quality;
        v.kid = id;
        v.value = vl;
        v.tm = tm;
        dq.push_back(v);
        u32_t day = (tm + BEIJING_TIME) / ONE_DAY_SECONDS * ONE_DAY_SECONDS;
        auto it = _kks_file.find(day);

        if (it == _kks_file.end()) {
            _kks_file[day] = std::shared_ptr<KKSFile>(new KKSFile());
            _kks_file[day]->attach(getKKSPath(day));
            _kks_file[day]->setMaxMemUsage(5LL * 1024 * 1024 * 1024); //5G
            it = _kks_file.find(day);
        }
        it->second->addKKS(id, tm, vl, quality);
    }
    if (++jjj % 1000 == 0) {
        //控制下清理频率
        TL_ThreadRwLock::WLock wr(_cache_write_read_lock);
        auto it = _kks_value_cache.begin();
        while (it != _kks_value_cache.end()) {
            auto &dqr = it->second;
            if (dqr.size() > 3) { //只保留最后1个值
                v = *dqr.rbegin();
                dqr.clear();
                dqr.push_back(v);
            }
            ++it;
        }
        //保留当天的数据
        while (_kks_file.size() > 1) {
            _kks_file.erase(_kks_file.begin());
        }
    }
}

/**
 * 获取kks，如果没有，则用指定路径创建
 * 导入数据的时候使用，其他情况下慎重使用
 * @param tm
 * @param kks_file_name
 * @return 
 */
std::shared_ptr<KKSFile> KKSManger::getKKSFile(u32_t tm, const string &kks_file_name) {
    std::lock_guard<std::mutex> guard(_mux);
    u32_t day = (tm + BEIJING_TIME) / ONE_DAY_SECONDS * ONE_DAY_SECONDS;
    if (_kks_file.find(day) == _kks_file.end()) {
        _kks_file[day] = std::shared_ptr<KKSFile>(new KKSFile());
        if (kks_file_name.empty()) {
            _kks_file[day]->attach(getKKSPath(day));
        } else {
            _kks_file[day]->attach(kks_file_name);
        }
    }
    return _kks_file[day];
}

/**
 * 获取当前正在使用的kks；
 * 返回的kks只能用于read
 * @return 
 */
std::shared_ptr<KKSFile> KKSManger::getCurrentKKSFile() {
    std::lock_guard<std::mutex> guard(_mux);
    if (_kks_file.size()) {
        return _kks_file.rbegin()->second;
    }
    return std::shared_ptr<KKSFile>(nullptr);
}

void KKSManger::kks2json(ostream &os, time_t tm, KKSFile::KKS_Value *v, bool float_fmt, char type) {
    //format
    //{ts:1526289939,type:'A',quality:'G',value:249.993}
    os << "{ts:" << tm << ",type:'" << type << "',quality:'" << v->quality << "',value:";
    if (float_fmt) {
        //KKSFile::KKS_Float_Value *fv = (KKSFile::KKS_Float_Value*)v;
        //os << *((float *) & v->value);
        os << ((KKSFile::KKS_Float_Value *) v)->value;
    } else {
        os << v->value;
    }
    os << "}";
}

void KKSManger::KKSLastValue(ostream &os, const vector<string> tgns) {
    //vector 
    std::shared_ptr<KKSFile> kfile = getCurrentKKSFile();
    u32_t kid = 0;
    string unfound_tgn;
    vector<char> rowbuff;
    KKSFile::KKS_ROW_Head *rhead;
    KKSFile::KKS_Value *value;
    auto it = tgns.begin();
    while (it != tgns.end()) {
        kid = _kks_index.getKKSId(*it);
        if (kid == 0) {
            if (unfound_tgn.empty()) unfound_tgn.append(",", 1);
            unfound_tgn.append(*it);
        }
        KKSFile::KKS_Meta *m = kfile->getMeta(kid);
        if (m) {
            off_t last_offset = m->offset_current;
            kfile->getRowData(last_offset, rowbuff);
            char *addr = &rowbuff[0];
            rhead = (KKSFile::KKS_ROW_Head *) addr;
            //rhead->tail_tm;//

            value = (KKSFile::KKS_Value *) (addr + sizeof(KKSFile::KKS_Value) * (rhead->count - 1));

        }
        ++it;
    }

}

void KKSManger::history(ostream &os, const string &tgn, time_t stime, time_t etime) {
    if (stime > etime) {
        os << "{\"ret\":-1,\"msg\":\"stime gt etime\"}";
        return;
    }

    u32_t kks_id = _kks_index.getKKSId(tgn);
    history(os, tgn, kks_id, stime, etime);
}

void KKSManger::history(ostream &os, const string &tgn, u32_t kks_id, time_t stime, time_t etime) {
    if (kks_id == 0) {
        os << "{\"ret\":-1,\"msg\":\"kks not found " << tgn << "\"}";
        return;
    }

    string kfile = getKKSPath(stime);
    //数据来源中含有时区
    //stime += BEIJIN_TIME;
    //etime += BEIJIN_TIME;

    if (!TL_Common::isFileExistEx(kfile)) {
        os << "{\"ret\":-1,\"msg\":\"kks file :" << kfile << " not exist.\"}";
        return;
    }
    //range : [stime,etime]
    KKSFile kks_file(kfile);
    KKSFile::KKS_Meta *meta = kks_file.getMeta(kks_id);
    if (meta == nullptr) {
        os << "{\"ret\":-1,\"msg\":\"kks meta not found for  :" << tgn << " [" << kks_id << "]\"}";
        return;
    }

    vector<off_t> rows;
    kks_file.getRowOffset(kks_id, stime, etime, rows);
    if (rows.size() == 0) {
        os << "{\"ret\":-1,\"retmsg\":\"kks[" << kks_id << "] has no data.\"}";
        return;
    }

    vector<char> buffer;

    KKSFile::KKS_ROW_Head *rhead = nullptr;
    KKSFile::KKS_Value *value = nullptr;
    //int last_i = rows.size() - 1;
    short ddtm = 0;
    u32_t last_tm = 0;
    int last_delta = 0;
    KKSFile::KKS_Float_Value *fvalue = nullptr;

    os << "{\"ret\":0,\"type\":\"A\",\"quality\":\"1\",\"tgn\":\"" << tgn << "\",";
    TL_Datetime dt(stime);
    os << "\"range\":\"" << dt.toString() << " | ";
    dt.setTime(etime);
    os << dt.toString() << "\",";
    os << "\"rows\":\"" << rows.size() << "#";
    int i = 0;
    for (; i < rows.size() - 1; ++i) {
        os << rows[i] << ",";
    }
    --i;
    if (i >= 0) {
        os << rows[i];
    }
    os << "\",\"stime\":" << stime << ",\"data\":[";

    //os << "]}";
    //return;
    ostringstream ostmp;
    for (int i = 0; i < rows.size(); ++i) {
        kks_file.getRowData(rows[i], buffer);
        char *addr = &buffer[0];
        rhead = (KKSFile::KKS_ROW_Head *) addr;
        value = (KKSFile::KKS_Value *) (addr + sizeof(KKSFile::KKS_ROW_Head));
        //ddtm = 0;
        last_delta = 0;
        last_tm = rhead->start_tm;

        for (u32_t j = 0; j < rhead->count; ++j) {
            fvalue = (KKSFile::KKS_Float_Value *) value;
            if (value->delta_delta_tm & 0x80) {
                KKSFile::KKS_Value *value0 = value;
                ++value;
                last_tm = value0->value;
                last_delta = 0;
                ddtm = 0;
                fvalue = (KKSFile::KKS_Float_Value *) value;
                ++j;
                //continue;
            } else if (value->delta_delta_tm & 0x40) {
                ddtm = -(value->delta_delta_tm & 0x3f);
            } else {
                ddtm = value->delta_delta_tm;
            }
            last_delta += ddtm; // + last_delta);
            last_tm += last_delta;
            //delta += ddtm;
            //++ii;
            //cerr << ii << ",(" << ddtm << "|" << last_tm << "~" << last_delta << "," << value->value << "|" << fvalue->value << ") " << endl;
            ++value;
            if (last_tm >= stime && last_tm <= etime) {
                ostmp << "[" << last_tm - stime << "," << fvalue->value << "],";
            } else if (last_tm > etime) {
                //last_tm > etime;提前结束循环
                break;
            }
        }
    }

//解决数据中最后一个逗号问题
    string data = ostmp.str();
    if (data.size() > 0) {
        data.erase(data.size() - 1);
    }
    os << data << "]}";
}

void KKSManger::getKKSHistory(u32_t kks_id, time_t stime, time_t etime, vector<KKSManger::KKS_Value> &values) {
    values.clear();
    if (stime > etime || kks_id == 0) {
        //cerr << __FILE__ << ":" << __LINE__ << "|" << stime << "|" << etime << "|" << kks_id << endl;
        return;
    }
    string kfile = getKKSPath(stime);
    //数据来源中含有时区
    //stime += BEIJIN_TIME;
    //etime += BEIJIN_TIME;
    if (!TL_Common::isFileExistEx(kfile)) {
        //cerr << __FILE__ << ":" << __LINE__ << "| data file not exist. " << kfile << endl;
        return;
    }
    //range : [stime,etime]
    std::shared_ptr<KKSFile> kks_file_ptr = getKKSFile(stime, kfile);
    //KKSFile kks_file(kfile);
    KKSFile::KKS_Meta *meta = kks_file_ptr->getMeta(kks_id);
    if (meta == nullptr) {
        //cerr << __FILE__ << ":" << __LINE__ << "|meta info is null:" << endl;
        return;
    }

    vector<off_t> rows;
    kks_file_ptr->getRowOffset(kks_id, stime, etime, rows);
    if (rows.size() == 0) {
        //cerr << __FILE__ << ":" << __LINE__ << "|not row offset for " << kks_id << "|" << stime << "|" << etime << endl;
        return;
    }

    vector<char> buffer;

    KKSFile::KKS_ROW_Head *rhead = nullptr;
    KKSFile::KKS_Value *value = nullptr;
    int last_i = rows.size() - 1;
    short ddtm = 0;
    u32_t last_tm = 0;
    int last_delta = 0;
    //KKSFile::KKS_Float_Value * fvalue = nullptr;
    //cerr << __FILE__ << ":" << __LINE__ << "|" <<  kks_id << "|" << stime << "|" << etime << "|rows:" << rows.size() << endl;
    KKS_Value real_value;
    for (int i = 0; i < rows.size(); ++i) {
        kks_file_ptr->getRowData(rows[i], buffer);
        char *addr = &buffer[0];
        rhead = (KKSFile::KKS_ROW_Head *) addr;
        value = (KKSFile::KKS_Value *) (addr + sizeof(KKSFile::KKS_ROW_Head));
        ddtm = 0;
        last_delta = 0;
        last_tm = rhead->start_tm;

        for (u32_t j = 0; j < rhead->count; ++j) {
            //fvalue = (KKSFile::KKS_Float_Value *)value;
            if (value->delta_delta_tm & 0x80) {
                KKSFile::KKS_Value *value0 = value;
                ++value;
                last_tm = value0->value;
                last_delta = 0;
                ddtm = 0;
                //fvalue = (KKSFile::KKS_Float_Value *)value;
                ++j;
            } else if (value->delta_delta_tm & 0x40) {
                ddtm = -(value->delta_delta_tm & 0x3f);
            } else {
                ddtm = value->delta_delta_tm;
            }
            last_delta += ddtm; // + last_delta);
            last_tm += last_delta;
            real_value.kid = kks_id;
            real_value.quality = value->quality;
            real_value.tm = last_tm;
            real_value.value = value->value;
            if (last_tm >= stime && last_tm <= etime) {
                values.push_back(real_value);
            }
            ++value;
        }
    }//end for
}

void KKSManger::getMultiKKSFromCache(ostream &os, const vector<string> &kks) {
    map<u32_t, KKSManger::KKS_Value> values;
    vector<u32_t> kids;
    auto it = kks.begin();
    while (it != kks.end()) {
        const string &s = *it;
        if (s.size() > 0)
            kids.push_back(atol(s.c_str()));
        ++it;
    }
    getMultiKKSFromCache(kids, values);
    int i = values.size();
    if (i > 0) {
        auto it = values.begin();
        os << "{\"ret\":0,\"ver\":\"2.0.0\",\"mod\":\"0.0.1\",\"msg\":\"\",\"rnd\":" << time(NULL)
           << ",\"rows\":" << i <<
           ",\"data\":{\"fields\":[\"tagId\",\"tagValue\",\"tagTime\"],\"values\":[";
        while (i > 1) {
            --i;
            const KKS_Value &kv = it->second;
            //const KKS_Float_Value * fv = (const KKS_Float_Value *)(&kv);
            os << "[" << kv.kid << "," << *(float *) (&kv.value) << "," << kv.tm << "],\n";
            ++it;
        }
        const KKS_Value &kv = it->second;
        os << "[" << kv.kid << "," << *(float *) (&kv.value) << "," << kv.tm << "] ]}}";
    } else {
        os << "{\"ret\":-1,\"ver\":\"2.0.0\",\"mod\":\"0.0.1\",\"msg\"=\"not found data.\"}";
    }
}

/**
 * 
 * @param kks
 * @param values
 * @return tm=0 基本无数据
 */
void KKSManger::getMultiKKSFromCache(const vector<string> &kks, map<string, KKSManger::KKS_Value> &values) {
    TL_ThreadRwLock::RLock rd(_cache_write_read_lock);
    values.clear();
    auto it = kks.begin();
    while (it != kks.end()) {
        const string &kname = *it;
        u32_t kid = _kks_index.getKKSId(kname);
        if (kid == 0) {
            values[kname].kid = kid;
        } else {
            auto it2 = _kks_value_cache.find(kid);
            if (it2 != _kks_value_cache.end() && it2->second.size() > 0) {
                values[kname] = *it2->second.rbegin();
            } else {
                values[kname].kid = kid;
            }
        }
        ++it;
    }
}

void KKSManger::getMultiKKSFromCache(const vector<u32_t> &kids, map<u32_t, KKSManger::KKS_Value> &values) {
    TL_ThreadRwLock::RLock rd(_cache_write_read_lock);
    values.clear();
    auto it = kids.begin();
    while (it != kids.end()) {
        u32_t kid = *it;
        if (kid == 0) {
            values[kid].kid = -1;
        } else {
            auto it2 = _kks_value_cache.find(kid);
            if (it2 != _kks_value_cache.end() && it2->second.size() > 0) {
                values[kid] = *it2->second.rbegin();
            } else {
                values[kid].kid = kid;
                //未找到，从数据文件中找最后一个数据
                //服务重启，导致cache丢失
            }
        }
        ++it;
    }
}

void KKSManger::getKKSValueIntMinute(ostream &os, const string &kks, int minute) {
    u32_t kid = _kks_index.getKKSId(kks);
    if (kid == 0) {
        os << "{ret:-1,retmsg:\"kks not found" << kks << "\"}";
        return;
    }
    time_t etime = time(NULL);
    if (minute <= 0) minute = 1;
    minute *= 60;
    time_t stime = etime - minute;
    string kfile = getKKSPath(etime);
    //数据来源中含有时区
    //stime += BEIJIN_TIME;
    //etime += BEIJIN_TIME;

    if (!TL_Common::isFileExistEx(kfile)) {
        os << "{ret:-1,retmsg:\"kks file :" << kfile << " not exist.\"}";
        return;
    }
    //range : [stime,etime]
    KKSFile kks_file(kfile);

    if (kid > 0) {
        vector<off_t> rows;
        kks_file.getRowOffset(kid, stime, etime, rows);
        if (rows.size() == 0) {
            os << "{ret:-1,retmsg:\"kks[" << kid << "] has no data.\"}";
            return;
        }

        vector<char> buffer;

        KKSFile::KKS_ROW_Head *rhead = nullptr;
        KKSFile::KKS_Value *value = nullptr;
        int last_i = rows.size() - 1;
        short ddtm = 0;
        u32_t last_tm = 0;
        int last_delta = 0;
        KKSFile::KKS_Float_Value *fvalue = nullptr;

        os << "{ret:0,tgn:\"" << kks << "\",";
        TL_Datetime dt(stime);
        os << "range:\"" << dt.toString() << " | ";
        dt.setTime(etime);
        os << dt.toString() << "\",";
        os << "rows:\"" << rows.size() << "#";
        for (int i = 0; i < rows.size(); ++i) {
            os << rows[i] << ",";
        }
        os << "\",stime:" << stime << ",data:[";

        //os << "]}";
        //return;
        ostringstream ostmp;
        for (int i = 0; i < rows.size(); ++i) {
            kks_file.getRowData(rows[i], buffer);
            char *addr = &buffer[0];
            rhead = (KKSFile::KKS_ROW_Head *) addr;
            value = (KKSFile::KKS_Value *) (addr + sizeof(KKSFile::KKS_ROW_Head));
            ddtm = 0;
            last_delta = 0;
            last_tm = rhead->start_tm;

            for (u32_t j = 0; j < rhead->count; ++j) {
                fvalue = (KKSFile::KKS_Float_Value *) value;
                if (value->delta_delta_tm & 0x80) {
                    KKSFile::KKS_Value *value0 = value;
                    ++value;
                    last_tm = value0->value;
                    last_delta = 0;
                    ddtm = 0;
                    fvalue = (KKSFile::KKS_Float_Value *) value;
                    //cerr << i << ",MatchPoint(" << value0->value << "," << value->value << ") " << endl;
                    //++value;
                    ++j;
                    //continue;
                } else if (value->delta_delta_tm & 0x40) {
                    ddtm = -(value->delta_delta_tm & 0x3f);
                } else {
                    ddtm = value->delta_delta_tm;
                }
                last_delta += ddtm; // + last_delta);
                last_tm += last_delta;
                //delta += ddtm;
                //++ii;
                //cerr << ii << ",(" << ddtm << "|" << last_tm << "~" << last_delta << "," << value->value << "|" << fvalue->value << ") " << endl;
                ++value;
                if (i == 0) {
                    //第一个row数据处理
                    if (last_tm >= stime) {
                        ostmp << "{ts:" << last_tm << ",type:'A',quality:'" << fvalue->quality << "',value:"
                              << fvalue->value << "},";
                    }
                } else if (i == last_i) {
                    //最后一个row处理
                    if (last_tm <= etime) {
                        ostmp << "{ts:" << last_tm << ",type:'A',quality:'" << fvalue->quality << "',value:"
                              << fvalue->value << "},";
                    } else {
                        //last_tm > etime;提前结束循环
                        break;
                    }
                } else {
                    //中间row
                    ostmp << "{ts:" << last_tm << ",type:'A',quality:'" << fvalue->quality << "',value:"
                          << fvalue->value << "},";
                }
            }
        }
        //解决数据中最后一个逗号问题
        string data = ostmp.str();
        if (data.size() > 0) data.erase(data.size() - 1, 1);
        os << data << "]}";
    }
}



